package com.relytosoft.mizar.media.host.emqx;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.relytosoft.mizar.media.common.constant.MediaMqConstants;
import com.relytosoft.mizar.media.common.emqx.client.MqttOp;
import com.relytosoft.mizar.media.external.api.system.request.DistrictDto;
import com.relytosoft.mizar.media.external.api.system.response.AddRegionDto;
import com.relytosoft.mizar.media.external.api.system.response.QueryRegionDto;
import com.relytosoft.mizar.media.external.api.system.response.RegionDto;
import com.relytosoft.mizar.media.external.api.system.service.RegionService;
import com.relytosoft.mizar.media.system.api.entity.SysDistrict;
import com.relytosoft.mizar.media.system.api.entity.SysRelatedPlatform;
import com.relytosoft.mizar.media.system.services.service.SysRelatedPlatformService;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @author Jierry
 * @version 2022/4/22.
 */
@Component
public class SystemMqttMessageListener implements com.relytosoft.mizar.media.common.emqx.listener.SystemMqttMessageListener {

    private static final Logger log = LoggerFactory.getLogger(SystemMqttMessageListener.class);

    @Resource
    private RegionService regionService;

    @Resource
    private SysRelatedPlatformService sysRelatedPlatformService;
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        if(topic.equals(MediaMqConstants.REGION_TOPIC)){
            boolean isConsume = false;
            String mqttMsg = new String(message.getPayload(), "utf-8");
            try {
                JSONObject jsonObject = JSON.parseObject(mqttMsg);
                String opType = jsonObject.getString("OpType");
                String data = jsonObject.getString("Data");
                switch (opType) {
                    //查询第三方根区域
                    case "QUERYROOT":
                        isConsume = queryRoot();
                        break;
                    //添加根区域至系统
                    case "INSERTROOT":
                        isConsume = insertRoot(data);
                        break;
                    //添加区域至第三方
                    case "INSERT":
                        isConsume = insert(data);
                        break;
                    //更新区域关联信息至系统
                    case "UPDATERELEATED":
                        isConsume = updateRelated(data);
                        break;
                    //查询第三方区域信息
                    case "QUERY":
                        isConsume = queryRegion(data);
                        break;
                    default:
                        break;
                }
                log.info("线程名称：【" + Thread.currentThread().getName() + "】" + "接收的主题：" + topic + ",接收的操作类型：" + opType + "，执行结果：" + isConsume);
            }
            catch (Exception e){
                log.info("线程名称：【" + Thread.currentThread().getName() + "】" + "接收的主题：" + topic + ",处理发生错误：" + e.getMessage());
            }
        }
    }
    /**
     * 查询第三方根区域，并发送添加根区域消息
     *
     * @return 消费成功
     */
    private boolean queryRoot() {
        //查询海康根区域
        RegionDto regionDTO = regionService.getRootRegion();
        if (regionDTO != null) {
            //发送添加根区域消息
            JSONObject jsonObject=new JSONObject();
            jsonObject.put("OpType","INSERTROOT");
            jsonObject.put("Data", regionDTO);
            MqttOp.publish(MediaMqConstants.REGION_TOPIC,JSON.toJSONString(jsonObject));
//            KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
//            kafkaOpMsg.setOpType("INSERTROOT");
//            kafkaOpMsg.setData(JSON.toJSONString(regionDTO));
//            kafkaProducer.send(MediaMqConstants.REGION_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
            return true;
        }
        return false;
    }
    /**
     * 添加根区域信息
     *
     * @param data 消息信息
     * @return 消费结果
     */
    private boolean insertRoot(String data) {
        //查询系统根区域信息
        SysRelatedPlatform sysRelatedPlatform = sysRelatedPlatformService.getRelatedPlatform(1, "HK", 0, 1);

        if (sysRelatedPlatform == null) {
            //不存在则添加
            RegionDto regionDTO = JSON.parseObject(data, RegionDto.class);
            sysRelatedPlatform = new SysRelatedPlatform();
            sysRelatedPlatform.setOwnId(0);
            sysRelatedPlatform.setType(1);
            sysRelatedPlatform.setPlatformName("HK");
            sysRelatedPlatform.setPlatformId(regionDTO.getIndexCode());
            sysRelatedPlatform.setIsSync(1);
            sysRelatedPlatform.setStatus(1);
            return sysRelatedPlatformService.save(sysRelatedPlatform);
        } else {
            //存在直接返回已消费
            return true;
        }
    }


    /**
     * 添加区域信息至第三方
     *
     * @param data 消息信息
     * @return 消费结果
     */
    private boolean insert(String data) {
        SysDistrict entity = JSON.parseObject(data, SysDistrict.class);
        //父级区域关联第三方标识为空
        if (StrUtil.isEmpty(entity.getParentPlatformId())) {
            //查询系统父级区域关联第三方
            SysRelatedPlatform sysRelatedPlatform = sysRelatedPlatformService.getRelatedPlatform(1, "HK", entity.getParentId(), 1);
            entity.setParentPlatformId(sysRelatedPlatform.getPlatformId());
        }
        if (!StrUtil.isEmpty(entity.getParentPlatformId())) {
            //父级区域已关联至第三方
            //调用第三方添加区域
            AddRegionDto addRegionDTO = regionService.addRegion(new DistrictDto(entity.getName(), entity.getRemark(), entity.getParentPlatformId()));
            if (addRegionDTO.getSuccesses() != null && addRegionDTO.getSuccesses().size() > 0) {
                //第三方添加区域成功
                //发送更新第三方区域信息到关联表的消息
                SysRelatedPlatform curRelatedPlatform = new SysRelatedPlatform();
                curRelatedPlatform.setId(entity.getRelatedPlatformId());
                curRelatedPlatform.setIsSync(1);
                curRelatedPlatform.setPlatformId(addRegionDTO.getSuccesses().get(0).getRegionIndexCode());
                curRelatedPlatform.setUpdatedBy(entity.getUpdatedBy());
                curRelatedPlatform.setUpdatedId(entity.getUpdatedId());
                JSONObject jsonObject=new JSONObject();
                jsonObject.put("OpType","UPDATERELEATED");
                jsonObject.put("Data", curRelatedPlatform);
                MqttOp.publish(MediaMqConstants.REGION_TOPIC,JSON.toJSONString(jsonObject));
//                KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
//                kafkaOpMsg.setOpType("UPDATERELEATED");
//                kafkaOpMsg.setData(JSON.toJSONString(curRelatedPlatform));
//                kafkaProducer.send(MediaMqConstants.REGION_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
                return true;
            } else {
                //第三方添加区域失败，发送查询区域消息
                JSONObject jsonObject=new JSONObject();
                jsonObject.put("OpType","QUERY");
                jsonObject.put("Data", entity);
                MqttOp.publish(MediaMqConstants.REGION_TOPIC,JSON.toJSONString(jsonObject));
//                KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
//                kafkaOpMsg.setOpType("QUERY");
//                kafkaOpMsg.setData(JSON.toJSONString(entity));
//                kafkaProducer.send(MediaMqConstants.REGION_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
                return true;
            }
        }
        return false;
    }


    /**
     * 更新第三方关联信息
     *
     * @param data 消息信息
     * @return 消费结果
     */
    private boolean updateRelated(String data) {
        SysRelatedPlatform curRelatedPlatform = JSON.parseObject(data, SysRelatedPlatform.class);
        return sysRelatedPlatformService.updateById(curRelatedPlatform);
    }

    /**
     * 查询第三方区域信息
     *
     * @param data 消息信息
     * @return 消费结果
     */
    private boolean queryRegion(String data) {
        SysDistrict entity = JSON.parseObject(data, SysDistrict.class);
        //查询第三方区域信息
        QueryRegionDto queryRegionDTO = regionService.searchRegion(new DistrictDto(entity.getName(), "", entity.getParentPlatformId()));

        if (queryRegionDTO.getTotal() > 0) {
            //更新第三方区域信息到关联表
            SysRelatedPlatform curRelatedPlatform = new SysRelatedPlatform();
            curRelatedPlatform.setId(entity.getRelatedPlatformId());
            curRelatedPlatform.setIsSync(1);
            curRelatedPlatform.setPlatformId(queryRegionDTO.getList().get(0).getIndexCode());
            curRelatedPlatform.setUpdatedBy(entity.getUpdatedBy());
            curRelatedPlatform.setUpdatedId(entity.getUpdatedId());

            JSONObject jsonObject=new JSONObject();
            jsonObject.put("OpType","UPDATERELEATED");
            jsonObject.put("Data", curRelatedPlatform);
            MqttOp.publish(MediaMqConstants.REGION_TOPIC,JSON.toJSONString(jsonObject));
//            KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
//            kafkaOpMsg.setOpType("UPDATERELEATED");
//            kafkaOpMsg.setData(JSON.toJSONString(curRelatedPlatform));
//            kafkaProducer.send(MediaMqConstants.REGION_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
        }
        return true;
    }
}
